/*
服务端主题的发布订阅实现
    1. 为dispatcher模块提供当主题请求到来时的回调函数
    2. 维护订阅者的管理信息
*/

#include "../Common/Net.hpp"
#include "unordered_set"

namespace RpcModule
{
    namespace ServerModule
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager() {}
            // 当有topic请求到来时，对主题的请求进行处理
            void OnTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                // 1. 对不同的主题请求做出处理
                TopicOptype optype = message->GetTopicOptype();
                bool ret;
                switch (optype)
                {
                // 主题的创建
                case TopicOptype::TOPIC_CREATE:
                    TopicCreate(conn, message);
                    break;
                // 主题的删除
                case TopicOptype::TOPIC_REMOVE:
                    TopicRemove(conn, message);
                    break;
                // 主题的订阅
                case TopicOptype::TOPIC_SUBSCRIBE:
                    ret = TopicSubscribe(conn, message);
                    break;
                // 主题的取消订阅
                case TopicOptype::TOPIC_CANCEL:
                    TopicUnSubscribe(conn, message);
                    break;
                // 主题消息的发布
                case TopicOptype::TOPIC_PUBLISH:
                    ret = TopicPublish(conn, message);
                    break;
                default:
                    return ErrorResponse(conn, message, RCode::RCODE_INVALID_OPTYPE);
                }
                // 2. 进行主题请求的响应
                if (!ret)
                    return ErrorResponse(conn, message, RCode::RCODE_NOT_FOUND_TOPIC);
                return NormalResponse(conn, message);
            }
            // 当有一个连接断开时，需要维护组织结构
            void OnShutDown(const BaseConnection::ptr &conn)
            {
                std::unordered_set<std::string> topics;
                SubScriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_sub = _subscribers.find(conn);
                    // 1. 消息发布者断开连接，什么也不用干
                    if (it_sub == _subscribers.end())
                        return;
                    // 2. 消息订阅者断开连接，需要删除该订阅者订阅的所有主题的关系
                    subscriber = it_sub->second;
                    topics = subscriber->topics;
                }
                // 1. 先维护_subscribers
                _subscribers.erase(conn);
                // 2. 再维护_topics
                for (auto &topic_name : topics)
                {
                    auto it_topic = _topics.find(topic_name);
                    if (it_topic == _topics.end())
                        continue;
                    it_topic->second->RemoveSubscriber(subscriber);
                }
            }

        private:
            // 当主题操作出错时的响应
            void ErrorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &message, RCode rcode)
            {
                TopicResponse::ptr response = MessageFactory::CreateMessage<TopicResponse>();
                response->SetId(message->GetRid());
                response->SetMType(MType::RSP_TOPIC);
                response->SetRcode(rcode);
                conn->Send(response);
            }
            // 正常处理主题操作的响应
            void NormalResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                TopicResponse::ptr response = MessageFactory::CreateMessage<TopicResponse>();
                response->SetId(message->GetRid());
                response->SetMType(MType::RSP_TOPIC);
                response->SetRcode(RCode::RCODE_OK);
                conn->Send(response);
            }
            // 主题的创建
            void TopicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                // 主题创建，没有订阅者，只需要创建主题，再维护topics即可
                // 1. 创建新的主题
                std::unique_lock<std::mutex> lock(_mutex);
                std::string topic_name = message->GetTopicKey();
                Topic::ptr newtopic = std::make_shared<Topic>(topic_name);
                // 2. 维护topics
                _topics.insert(std::make_pair(topic_name, newtopic));
            }
            // 主题的删除
            void TopicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                // 1. 判断是否有该主题
                std::string topic_name = message->GetTopicKey();
                std::unordered_set<SubScriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                        return;
                    // 2. 先删除name主题的主题信息
                    Topic::ptr topic = it->second;
                    subscribers = topic->subscribers;
                    _topics.erase(it);
                }
                // 3. 再删除主题的订阅者信息
                for (auto &subscriber : subscribers)
                {
                    subscriber->RemoveTopic(topic_name);
                }
            }
            // 主题的订阅
            bool TopicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                // 1. 先查找主题，看要订阅的主题是否存在
                std::string topic_name = message->GetTopicKey();
                Topic::ptr topic;
                SubScriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                    {
                        LOGERROR("不存在%s主题，无法订阅不存在的主题", topic_name.c_str());
                        return false;
                    }
                    topic = it->second;
                    // 2. 走到这里，说明存在主题
                    // 查找是否有主题订阅者，没有创建，有则订阅
                    auto it_sub = _subscribers.find(conn);
                    if (it_sub == _subscribers.end())
                    {
                        // 不存在主题订阅者，创建主题订阅者
                        subscriber = std::make_shared<SubScriber>(conn);
                        _subscribers.insert(std::make_pair(conn, subscriber));
                    }
                    else
                    {
                        subscriber = it_sub->second;
                    }
                }
                // 这里需要特别注意：下面的两个操作需要在锁之外进行调用
                // 因为他们内部都各自有锁，所以需要防止死锁
                // 3. 有了主题订阅者，进行主题的订阅
                subscriber->AppendTopic(topic_name);
                // 4. 一个主题新增了一个订阅者
                topic->AppendSubscriber(subscriber);
                return true;
            }
            // 主题的取消订阅
            void TopicUnSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                // 1. 先查看主题是否存在
                Topic::ptr topic;
                SubScriber::ptr subscriber;
                std::string topic_name = message->GetTopicKey();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics.find(topic_name);
                    // 2. 主题存在，判断conn连接是否有主题订阅者
                    if (it_topic != _topics.end())
                        topic = it_topic->second;
                    auto it_sub = _subscribers.find(conn);
                    if (it_sub != _subscribers.end())
                        subscriber = it_sub->second;
                }
                // 3. 进行主题的取消订阅
                if (subscriber)
                    subscriber->RemoveTopic(topic_name);
                // 4. 需要删除主题与conn的关系
                if (topic && subscriber)
                    topic->RemoveSubscriber(subscriber);
            }
            // 主题消息的发布
            bool TopicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &message)
            {
                // 1. 先查看主题是否存在
                Topic::ptr topic;
                SubScriber::ptr subscriber;
                std::string topic_name = message->GetTopicKey();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it_topic = _topics.find(topic_name);
                    // 2. 主题存在，判断conn连接是否有主题订阅者
                    if (it_topic == _topics.end())
                    {
                        LOGERROR("不存在%s主题，无法进行消息发布", topic_name.c_str());
                        return false;
                    }
                    topic = it_topic->second;
                }
                // 3. 进行主题消息的发布
                topic->PushMessage(message);
                return true;
            }

        private:
            // 主题订阅者的管理
            struct SubScriber
            {
                using ptr = std::shared_ptr<SubScriber>;
                std::mutex mutex;
                BaseConnection::ptr conn;
                // 一个主题订阅者订阅了多少主题
                std::unordered_set<std::string> topics;

                SubScriber(const BaseConnection::ptr &connection) : conn(connection) {}
                // 主题的订阅
                void AppendTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.insert(topic_name);
                }
                // 主题的取消订阅
                void RemoveTopic(const std::string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    topics.erase(topic_name);
                }
            };
            // 针对主题的管理
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex mutex;
                std::string topic_name;
                // 一个主题有多少订阅者
                std::unordered_set<SubScriber::ptr> subscribers;

                Topic(const std::string &name) : topic_name(name) {}
                // 获取主题名称
                std::string GetTopicName() { return topic_name; }
                // 获取该主题的所有订阅者
                std::unordered_set<SubScriber::ptr> GetSubScribers() { return subscribers; }
                // 新增主题订阅者 -- 当客户端进行主题订阅时
                void AppendSubscriber(const SubScriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    subscribers.insert(subscriber);
                }
                // 删除主题订阅者 -- 当客户端断开连接时
                void RemoveSubscriber(const SubScriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    subscribers.erase(subscriber);
                }
                // 消息的发布功能 -- 查找_subscribers，进行消息的推送
                void PushMessage(const BaseMessage::ptr &message)
                {
                    std::unique_lock<std::mutex> lock(mutex);
                    for (auto &subscriber : subscribers)
                    {
                        subscriber->conn->Send(message);
                    }
                }
            };

        private:
            std::mutex _mutex;
            // 一个主题被哪些订阅者订阅了 -- 用于主题消息的推送
            std::unordered_map<std::string, Topic::ptr> _topics;
            // 一个连接有什么主题订阅者 -- 用于连接关闭，进行_topics中的删除
            std::unordered_map<BaseConnection::ptr, SubScriber::ptr> _subscribers;
        };
    }
}